package net.ib.rpc.handler;

import static org.jboss.netty.channel.Channels.write;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import net.ib.mn.protocol.CommonEnvelope.Envelope;
import net.ib.mn.protocol.CommonEnvelope.EnvelopeType;
import net.ib.rpc.message.RpcRequestInfo;
import net.ib.rpc.message.RpcResponseInfo;
import net.ib.rpc.method.MethodInfo;
import net.ib.rpc.method.MethodInfos;

import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;

import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import com.google.protobuf.RpcCallback;

public class RpcHandler extends SimpleChannelHandler {
	private final MethodInfos localMethods;
	private final MethodInfos remoteMethods;

	private final ConcurrentMap<Integer, RpcResponseInfo> pendingResInfos = new ConcurrentHashMap<Integer, RpcResponseInfo>();
	private AtomicInteger prevRequestId = new AtomicInteger(0);

	public RpcHandler(MethodInfos localServices, MethodInfos remoteServices) {
		this.localMethods = localServices;
		this.remoteMethods = remoteServices;
	}

	/**
	 * Input Data Processing
	 * 
	 * 직접 요청하는 것인지, 요청에 대한 응답인지를 구분 해서 처리한다.
	 * 
	 * - 기본적으로 ChannelHandler 에서는 upstream을 호출한다.
	 */
	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		// GPB Envelope인지 확인한다.
		Object message = e.getMessage();
		if (!(message instanceof Envelope)) {
			super.messageReceived(ctx, e);
		}

		Envelope envelope = (Envelope) message;

		if (envelope.getType() == EnvelopeType.REQUEST) {
			// 요청 받은 메시지
			MethodInfo methodInfo = localMethods.getMethodInfo(envelope
					.getMethodId());
			processRequestReceived(ctx, envelope, methodInfo);
		} else if (envelope.getType() == EnvelopeType.RESPONSE) {
			// 응답 받은 메시지
			MethodInfo entry = remoteMethods.getMethodInfo(envelope
					.getMethodId());
			processResponseReceived(ctx, envelope, entry);
		} else {
			// 정의 되지 않은 메시지
			super.messageReceived(ctx, e);
		}
	}

	private void processRequestReceived(final ChannelHandlerContext ctx,
			Envelope envelope, MethodInfo methodInfo)
			throws InvalidProtocolBufferException {
		if (methodInfo == null) {
			return;
			// Method 존재하지 않음!
			// logger.error("Method is not existed. method id:{}",
			// String.format("0x%08X", envelope.getMethodId()));
		}

		Message message = methodInfo.getService()
				.getRequestPrototype(methodInfo.getMethod())
				.newBuilderForType().mergeFrom(envelope.getBody()).build();

		methodInfo.getService()
				.callMethod(methodInfo.getMethod(), null, message,
						new WriteResponse<Message>(ctx.getChannel(), envelope));
	}

	private void processResponseReceived(ChannelHandlerContext ctx,
			Envelope envelope, MethodInfo entry)
			throws InvalidProtocolBufferException {
		RpcResponseInfo responseInfo = pendingResInfos.remove(envelope
				.getRequestId());
		if (responseInfo == null)
			return;
		Message message = responseInfo.getResponseType().newBuilderForType()
				.mergeFrom(envelope.getBody()).build();

		if (responseInfo.getCallback() == null)
			return;

		responseInfo.getCallback().run(message);
	}

	/**
	 * Output Data Processing
	 * 
	 * 직접 요청하는 것인지, 요청에 대한 응답인지를 구분 해서 처리한다.
	 * 
	 * - 기본적으로 ChannelHandler 에서는 downsteam 을 호출한다.
	 */
	@Override
	public void writeRequested(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		Object message = e.getMessage();
		if (!(message instanceof RpcRequestInfo)) {
			super.writeRequested(ctx, e);
		}

		RpcRequestInfo requestInfo = (RpcRequestInfo) message;
		if (requestInfo.getType() == EnvelopeType.REQUEST) {
			processRequestWriteRequested(ctx, e, requestInfo);
		} else if (requestInfo.getType() == EnvelopeType.RESPONSE) {
			processResponseWriteRequested(ctx, e, requestInfo);
		} else {
			super.writeRequested(ctx, e);
		}
	}

	private void processRequestWriteRequested(final ChannelHandlerContext ctx,
			MessageEvent e, RpcRequestInfo requestInfo) throws Exception {

		int requestId = prevRequestId.incrementAndGet();
		Envelope envelope = Envelope
				.newBuilder()
				.setBody(requestInfo.getBody().toByteString())
				.setMethodId(remoteMethods.getMethodId(requestInfo.getMethod()))
				.setRequestId(requestId).setType(requestInfo.getType()).build();

		RpcResponseInfo rpcResInfo = requestInfo.getResponseInfo();
		rpcResInfo.setRequestId(requestId);
		pendingResInfos.put(requestId, rpcResInfo);
		write(ctx, e.getFuture(), envelope, null);
	}

	private void processResponseWriteRequested(ChannelHandlerContext ctx,
			MessageEvent e, RpcRequestInfo requestInfo) {

		Envelope envelope = Envelope.newBuilder()
				.setBody(requestInfo.getBody().toByteString())
				.setMethodId(requestInfo.getMethodId())
				.setRequestId(requestInfo.getRequestId())
				.setType(requestInfo.getType()).build();

		write(ctx, e.getFuture(), envelope, null);
	}

	/**
	 * Rpc 요청에 대한 응답 콜백
	 * 
	 * GPB Service Impl 에서 response 구현해서 리턴 해야함.
	 * 
	 * @comment happy2v
	 * 
	 * @param <T>
	 */
	private class WriteResponse<T extends Message> implements RpcCallback<T> {

		private final Channel channel;
		private final int requestId;
		private final int methodId;

		public WriteResponse(Channel channel, Envelope requestEnvelope) {
			this.channel = channel;
			this.requestId = requestEnvelope.getRequestId();
			this.methodId = requestEnvelope.getMethodId();
		}

		@Override
		public void run(T response) {
			if (!channel.isConnected()) {
				return;
			}

			RpcRequestInfo requestInfo = new RpcRequestInfo(response,
					EnvelopeType.RESPONSE, null, null, requestId, methodId,
					Long.MAX_VALUE);
			channel.write(requestInfo);
		}
	}
}
